package bootsxt.utils;

import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import java.util.logging.Logger;

/**
 * Kafka工具类，支持生产和订阅消息，支持多个消费者组，每个消费者组独立消费。
 */
public class Kafka {



    // 存储所有topic
    private static final ConcurrentHashMap<String, Topic<?>> topics = new ConcurrentHashMap<>();
    private static final Logger logger = Logger.getLogger(Kafka.class.getName());
    private static final org.slf4j.Logger log = LoggerFactory.getLogger(Kafka.class);

    // 私有构造函数，防止外部实例化
    private Kafka() {}

    /**
     * 获取或创建一个Topic实例。
     *
     * @param topicName 主题名称
     * @param msgType 消息类型
     * @param clazz 消息的类型
     * @param <T> 消息类型的泛型
     * @return Topic实例
     */
    public static <T> Topic<T> topic(String topicName, String msgType, Class<T> clazz) {
        String fullTopicName = msgType.concat("/").concat(topicName);
        return getTopic(fullTopicName, clazz);
    }

    public static ConcurrentHashMap<String, Topic<?>> getTopics() {
        return topics;
    }

    /**
     * 获取Topic实例。
     *
     * @param name 主题全名
     * @param clazz 消息类型
     * @param <T> 消息类型的泛型
     * @return Topic实例
     */
    @SuppressWarnings("unchecked")
    private static <T> Topic<T> getTopic(String name, Class<T> clazz) {
        return (Topic<T>) topics.computeIfAbsent(name, n -> new Topic<>(n));
    }

    /**
     * Topic类：管理消息的生产和消费，支持多个消费者组。
     *
     * @param <T> 消息类型的泛型
     */
    public static class Topic<T> {
        private final String name;
        private final Map<String,Function<T,Void>> scribes = new ConcurrentHashMap<>();

        private final List<Object> pingMsgStatus = new CopyOnWriteArrayList<>();
        private Topic(String name) {
            this.name = name;
        }

        public boolean pingMsg(long timeoutMs) throws InterruptedException {
            Object isMs = new Object();
            pingMsgStatus.add(isMs);
            Thread.sleep(timeoutMs);
            return ! pingMsgStatus.contains(isMs);
        }

        public synchronized void unsubscribe(String id) {
            if (scribes.containsKey(id)){
                log.info("start unsubscribing topic: {}  scribes size {}  id {}", name,scribes.size(),id);
                scribes.remove(id);
                log.info("end unsubscribing topic: {}  scribes size {}  id {}", name,scribes.size(),id);
            }
        }

        public synchronized boolean isScribe(String id) {
            return scribes.containsKey(id);
        }

        public Map<String, Function<T, Void>>  getScribes() {
            return scribes;
        }

        public static class Runnable implements Function<String, Void> {
            @Override
            public Void apply(String s) {
                return null;
            }
        }


        public void scribes(Function<T,Void> runnable,String id) {
            log.info("start scribes topic: {}  scribes size {} id {}", name,scribes.size(),id);
            scribes.put(id,runnable);
            log.info("end scribes topic: {}  scribes size {} id {}", name,scribes.size(),id);

        }



        /**
         * 生产消息到主题。
         *
         * @param messageContent 消息内容
         */
        public void produce(T messageContent) {
            pingMsgStatus.clear();
            scribes.keySet().stream().parallel().forEach(key -> {
                try {
                    scribes.get(key).apply(messageContent);
                }catch (Throwable e){
                    e.printStackTrace();
                    unsubscribe(key);
                }
            });
           // logger.info("Produced message to topic " + name + ". Total messages: " + messages.size());
        }

        /**
         * 获取主题名称。
         *
         * @return 主题名称
         */
        public String getName() {
            return name;
        }

        @Override
        public String toString() {
            return "Topic{" +
                    "name='" + name +
                    '}';
        }
    }
}
